package com.sunl.mall.ali.mq.producer;

import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.sunl.mall.ali.mq.api.producer.RocketMqProducer;
import com.sunl.mall.ali.mq.model.RocketMqMessageWrapper;
import com.sunl.mall.ali.mq.model.RocketMqSendResult;
import com.sunl.mall.ali.mq.serializer.RocketMqSerializer;
import com.sunl.mall.common.api.util.ExceptionUtil;
import com.sunl.mall.common.api.util.SystemClock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class RocketMqProducerBean extends AbstractRocketMqProducer implements RocketMqProducer {
    private final static Logger log = LoggerFactory.getLogger(RocketMqProducerBean.class);

    /**
     * 需要注入该字段，指定构造{@link Producer}实例的属性，具体支持的属性详见{@link PropertyKeyConst}
     */
    private Properties properties;

    public void setProperties(Properties properties) {
        this.properties = properties;
    }

    private Producer producer;

    public RocketMqProducerBean(){}

    @Override
    public <T> boolean send(RocketMqMessageWrapper<T> message, RocketMqSerializer<T> serializer) {
        return sendBackResult(message, serializer).isSuccess();
    }

    @Override
    public <T> RocketMqSendResult sendBackResult(RocketMqMessageWrapper<T> message, RocketMqSerializer<T> serializer) {
        if (message.getContent() == null) {
            Object[] params = {message.getTopic(), message.getTag(), message.getKey()};
            log.error("Mq send Failure, Because Message content is null , topic:[{}], tag:[{}], key:[{}]", params);
            return RocketMqSendResult.fail();
        }
        SendResult sendResult = null;
        long start = SystemClock.currentTimeMillis();
        try {
            byte[] bytes = serializer.serialize(message.getContent());
            message.setBody(bytes);
            if (!checkSendMsgBody(message)) {
                return RocketMqSendResult.fail();
            }
            sendResult = this.producer.send(message);
        } catch (Throwable e) {
            Object[] params = {message.getTopic(), message.getTag(), message.getKey(), message.getContent(), SystemClock.currentTimeMillis() - start, ExceptionUtil.getAsString(e)};
            log.error("Mq send Exception, topic:[{}], tag:[{}], key:[{}], message:[{}], costTime:{}ms, Some Exception Occur:[{}]", params);
        }
        return RocketMqSendResult.successIfNotNull(sendResult);
    }

    @Override
    public <T> boolean sendOneway(RocketMqMessageWrapper<T> message, RocketMqSerializer<T> serializer) {
        if (message.getContent() == null) {
            Object[] params = {message.getTopic(), message.getTag(), message.getKey()};
            log.error("Mq sendOneway Failure, Because Message content is null , topic:[{}], tag:[{}], key:[{}]", params);
            return false;
        }
        boolean sendSuccess = true;
        long start = SystemClock.currentTimeMillis();
        try {
            byte[] bytes = serializer.serialize(message.getContent());
            message.setBody(bytes);
            if (!checkSendMsgBody(message)) {
                return false;
            }
            this.producer.sendOneway(message);
        } catch (Throwable e) {
            sendSuccess = false;
            Object[] params = {message.getTopic(), message.getTag(), message.getKey(), message.getContent(), SystemClock.currentTimeMillis() - start, ExceptionUtil.getAsString(e)};
            log.error("Mq send Exception(Oneway), topic:[{}], tag:[{}], key:[{}], message:[{}], costTime:{}ms, Some Exception Occur:[{}]", params);
        }
        return sendSuccess;
    }

    @Override
    public <T> boolean sendAsync(RocketMqMessageWrapper<T> message, RocketMqSerializer<T> serializer, SendCallback callback) {
        if (message.getContent() == null) {
            Object[] params = {message.getTopic(), message.getTag(), message.getKey()};
            log.error("Mq sendAsync Failure, Because Message content is null , topic:[{}], tag:[{}], key:[{}]", params);
            return false;
        }
        boolean sendSuccess = true;
        long start = SystemClock.currentTimeMillis();
        try {
            byte[] bytes = serializer.serialize(message.getContent());
            message.setBody(bytes);
            if (!checkSendMsgBody(message)) {
                return false;
            }
            this.producer.sendAsync(message, callback);
        } catch (Throwable e) {
            sendSuccess = false;
            Object[] params = {message.getTopic(), message.getTag(), message.getKey(), message.getContent(), SystemClock.currentTimeMillis() - start, ExceptionUtil.getAsString(e)};
            log.error("Mq send Exception(Async), topic:[{}], tag:[{}], key:[{}], message:[{}], costTime:{}ms, Some Exception Occur:[{}]", params);
        }
        return sendSuccess;
    }

    @Override
    public boolean isStarted() {
        return this.producer.isStarted();
    }

    @Override
    public boolean isClosed() {
        return this.producer.isClosed();
    }

    @Override
    public void start() {
        if (null == this.properties) {
            throw new ONSClientException("properties not set");
        }
        init(this.properties);
        this.producer = ONSFactory.createProducer(this.properties);
        this.producer.start();
    }

    @Override
    public void updateCredential(Properties credentialProperties) {
        if (this.producer != null) {
            this.producer.updateCredential(credentialProperties);
        }
    }

    @Override
    public void shutdown() {
        if (this.producer != null) {
            this.producer.shutdown();
        }
    }
}
